
/*
 * Hlpay-Plus aggregate payment system.
 * Copyright (c) 2024-2025 Hlpay Team Copyright has the right of final interpretation.
 */

package com.hlkj.pay.listener;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import com.hlkj.pay.app.merchant.IMerchantReportAppService;
import com.hlkj.pay.common.constants.MqConstant;
import com.hlkj.pay.dto.merchant.report.MerchantReportConsumerDto;

import lombok.extern.slf4j.Slf4j;

/**
 * @author monk@techsun.com
 * @date 2023/03/02 9:49
 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "rocketmq", name = "enabled", havingValue = "true")
@RocketMQMessageListener(topic = MqConstant.MERCHANT_REPORT_STATUS_TOPIC, consumerGroup = MqConstant.MERCHANT_REPORT_STATUS_TOPIC)
public class MerchantReportConsumer implements RocketMQListener<MerchantReportConsumerDto>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private IMerchantReportAppService merchantReportAppService;
    @Value("${server.port}")
    private String serverPort;
    @Override
    public void onMessage(MerchantReportConsumerDto merchantReportConsumerDto) {
        // 处理消息的逻辑
        // System.out.println("Received merchantReportConsumerDto: " + merchantReportConsumerDto);
        // Map<String, Object> params = JsonUtils.parseObject(merchantReportConsumerDto, Map.class);
        merchantReportAppService.processAccessMessage(merchantReportConsumerDto);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 设置当前实例的名称
        consumer.setInstanceName(MqConstant.MERCHANT_REPORT_STATUS_TOPIC + "_" + serverPort);
    }
}
